package com.myCompany.edu.realtime.app.dws;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.myCompany.edu.realtime.app.BaseAppV1;
import com.myCompany.edu.realtime.bean.InteractionCourseUserReviewBean;
import com.myCompany.edu.realtime.common.Constant;
import com.myCompany.edu.realtime.util.DateUtils;
import com.myCompany.edu.realtime.util.DimUtil;
import com.myCompany.edu.realtime.util.DruidDSUtil;
import com.myCompany.edu.realtime.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.time.Duration;

/*
    互动域课程粒度评价各窗口汇总表
    课程总评分、课程评价总人数、课程好评数，课程总评数


 */
public class Dws09_DwsInteractionCourseUserReviewWindow extends BaseAppV1 {

    public static void main(String[] args) {
        new Dws09_DwsInteractionCourseUserReviewWindow().init(
                3009,2,"Dws09_DwsInteractionCourseUserReviewWindow", Constant.TOPIC_DWD_INTERACTION_REVIEW);
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        // 1. 封装数据成bean
        // 注：一个用户只会一次评论
        SingleOutputStreamOperator<InteractionCourseUserReviewBean> beanStream = parseToBean(stream);
//        beanStream.print();

        // 2. 开窗聚合度量值
        SingleOutputStreamOperator<InteractionCourseUserReviewBean> winAndAggStream = winAndAgg(beanStream);

//        winAndAggStream.print();

        // 3. 补充维表数据 courseName
        SingleOutputStreamOperator<InteractionCourseUserReviewBean> resultStream = addDims(winAndAggStream);
//        resultStream.print();
        // 4.写出到clickhouse
        writeToClickHouse(resultStream);


    }

    private void writeToClickHouse(SingleOutputStreamOperator<InteractionCourseUserReviewBean> resultStream) {
        resultStream.addSink(
                FlinkSinkUtil.getClickHouseSink("dws_interaction_course_user_review_window",InteractionCourseUserReviewBean.class));
    }

    private SingleOutputStreamOperator<InteractionCourseUserReviewBean> addDims(SingleOutputStreamOperator<InteractionCourseUserReviewBean> stream) {
        return stream.map(new RichMapFunction<InteractionCourseUserReviewBean, InteractionCourseUserReviewBean>() {

            private Connection conn;

            @Override
            public void open(Configuration parameters) throws Exception {
                // 获取hbase（phoenix连接）
                DruidDataSource dataSource = DruidDSUtil.getDataSource();
                conn = dataSource.getConnection();
            }

            @Override
            public void close() throws Exception {
                if (conn != null){
                    conn.close();
                }
            }

            @Override
            public InteractionCourseUserReviewBean map(InteractionCourseUserReviewBean bean) throws Exception {
                // 读取维度表信息
                JSONObject courseInfo = DimUtil.readDimFromPhoenix(conn, "dim_course_info", bean.getCourseId());

                bean.setCourseName(courseInfo.getString("COURSE_NAME"));
                return bean;
            }
        });


    }

    private SingleOutputStreamOperator<InteractionCourseUserReviewBean> winAndAgg(SingleOutputStreamOperator<InteractionCourseUserReviewBean> beanStream) {
        return beanStream
                .assignTimestampsAndWatermarks(
                WatermarkStrategy.<InteractionCourseUserReviewBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner((bean,ts)->bean.getTs()))
                .keyBy(InteractionCourseUserReviewBean::getCourseId)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<InteractionCourseUserReviewBean>() {
                            @Override
                            public InteractionCourseUserReviewBean reduce(InteractionCourseUserReviewBean value1,
                                                                          InteractionCourseUserReviewBean value2) throws Exception {
                                value1.setMarkSum(value1.getMarkSum() + value2.getMarkSum());
                                value1.setUserReCt(value1.getUserReCt() + value2.getUserReCt());
                                value1.setGoodReCt(value1.getGoodReCt() + value2.getGoodReCt());
                                value1.setReCt(value1.getReCt() + value2.getReCt());
//
                                return value1;
                            }
                        },
                        new ProcessWindowFunction<InteractionCourseUserReviewBean, InteractionCourseUserReviewBean, String, TimeWindow>() {
                            @Override
                            public void process(String courseId, Context context, Iterable<InteractionCourseUserReviewBean> iterable, Collector<InteractionCourseUserReviewBean> out) throws Exception {
                                InteractionCourseUserReviewBean bean = iterable.iterator().next();
                                bean.setStt(DateUtils.toDateTime(context.window().getStart()));
                                bean.setEdt(DateUtils.toDateTime(context.window().getEnd()));

                                bean.setTs(System.currentTimeMillis());

                                out.collect(bean);
                            }
                        });
//
    }

    private SingleOutputStreamOperator<InteractionCourseUserReviewBean> parseToBean(DataStreamSource<String> stream) {
//        stream.print();
        return stream
                .map(new RichMapFunction<String, InteractionCourseUserReviewBean>() {
                    @Override
                    public InteractionCourseUserReviewBean map(String value) throws Exception {
                        JSONObject data = JSON.parseObject(value);

                        String courseId = data.getString("course_id");
                        Long markSum = data.getLong("review_stars");
                        Long ts = data.getLong("ts") * 1000;

                        Long goodRect = 0L;//好评数
                        if (markSum == 5){
                          goodRect = 1l;
                        }

                        return new InteractionCourseUserReviewBean(
                                "","",
                                courseId,"",markSum,1l,goodRect,1l,ts);
                    }
                });
    }


}
